package org.wolfengi.handler.mapred;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.wolfengi.handler.entity.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wolfengi.handler.entity.*;

import java.io.IOException;

/**
 * @Author: wolfengi
 * @Description: ${TODO}
 * @Date: Created in 10:27 2020/5/18
 * @Modified By:
 */
public class FaultReducer {

    private static final Logger log = LoggerFactory.getLogger(FaultReducer.class);

    public static class ShipTypeCountReduce extends Reducer<Text, IntWritable, TableOperForShipType, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipTypeCountReduce.");
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            String[] split = key.toString().split("@");
            String mmsi = split[1].trim();
            String syear = split[0].trim();
            String fault = split[2].trim();
            context.write(new TableOperForShipType(mmsi, syear, fault, count), NullWritable.get());
        }
    }

    public static class ShipFaultCountReduce extends Reducer<Text, IntWritable, TableOperForShipFalut, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipFaultCountReduce.");
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            context.write(new TableOperForShipFalut(key.toString().trim(), count), NullWritable.get());
        }
    }

    public static class ShipTravelRateCountReduce extends Reducer<Text, IntWritable, TableOperForShipTravelRate, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipTravelRateCountReduce.");
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            String[] split = key.toString().split("@");
            String mmsi = split[0].trim();
            String datetime = split[1].trim();
            String state = split[2].trim();
            context.write(new TableOperForShipTravelRate(mmsi, datetime, state, count), NullWritable.get());
        }
    }

    //未真正实现
    public static class ShipAgeCountReduce extends Reducer<Text, IntWritable, TableOperForShipAge, IntWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipAgeCountReduce.");
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            context.write(new TableOperForShipAge(key.toString().trim(), count), new IntWritable(count));
        }
    }

    public static class ShipFaultTypeCountReduce extends Reducer<Text, IntWritable, TableOperForShipFalutType, IntWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipFaultTypeCountReduce.");
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            String[] split = key.toString().split("@");
            context.write(new TableOperForShipFalutType(split[0].trim(), split[1].trim(), count), new IntWritable(count));
        }
    }

    public static class ShipGroupCountReduce extends Reducer<Text, IntWritable, TableOperForShipGroup, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipGroupCountReduce.");
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            context.write(new TableOperForShipGroup(key.toString().trim(), count), NullWritable.get());
        }
    }

    public static class ShipOverviewReduce extends Reducer<Text, IntWritable, TableOperForShipOverview, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipOverviewReduce.");
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            context.write(new TableOperForShipOverview(key.toString().trim(), count), NullWritable.get());
        }
    }

    public static class ShipDictReduce extends Reducer<Text, IntWritable, TableOperForShipDict, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            log.info("[-] executing ShipOverviewReduce.");
            String[] split = key.toString().split("@");
            if (split.length >= 4) {
                String mmsi = split[0].trim();
                float weight = Float.parseFloat(split[1].trim());
                int age = Integer.parseInt(split[2].trim());
                String type = split[3].trim();
                context.write(new TableOperForShipDict(mmsi, weight, age, type), NullWritable.get());
            }

        }

    }

    public static class ShipAgeBakReduce extends Reducer<Text, Text, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            String age = null;
            int count = 0;
            for (Text value : values) {
                if (key.toString().equals(value.toString())) {
                    count += 1;
                } else {
                    age = value.toString();
                }
            }
            int i = 0;
            if (null != age) {
                i = Integer.parseInt(age);
            }
            if (count > 0) {
                if (i > 0 && i <= 5) {
                    context.write(key, new Text(age + "=0-5=" + String.valueOf(count)));
                } else if (i > 5 && i <= 10) {
                    context.write(key, new Text(age + "=5-10=" + String.valueOf(count)));
                } else if (i > 10 && i <= 15) {
                    context.write(key, new Text(age + "=6-15=" + String.valueOf(count)));
                } else if (i > 15) {
                    context.write(key, new Text(age + "=>15=" + String.valueOf(count)));
                }
            }

        }
    }
}
